package com.yz.kafka;


import com.yz.pojo.KafkaConstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @author wlf
 * @data: 2021-10-18
 * @Description kafka监听
 */
@Slf4j
@Component
public class Listener {

    @KafkaListener(groupId = KafkaConstants.TOPIC_WYZ, topics = KafkaConstants.WYZ,id = "wyz-1")
    public void process(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {

       try {
           System.out.println(KafkaConstants.GROUP_ID_WYZ+">>>>>>>>>>>>>topic = {"+ record.topic()+"}, offset = {"+ record.offset()+"}, value = {"+record.value()+"} ");
           acknowledgment.acknowledge();
       }catch (Exception e){
           e.printStackTrace();
       }

    }

    /**
     * 通过采集平台采集到数据后 会分发到两个kafka Tiopic中
     * 我通过监听两个Topic 来消费数据 存储到不同的数据库中,普通数据 存储到TD数据库 并且调用device的feign接口更新缓存
     * 事件数据,存储到mongodb数据库中
     *  可能会问的问题: 怎么保证数据不丢失
     *      broker数据不丢失  一个集群肯定有多个broker,只要数据写到一个broker里面他肯定会自动同步
     *                      生产者通过分区的leader写入数据后，所有在ISR中follower都会从leader中复制数据，这样，可以确保即使leader崩溃了，其他的follower的数据仍然是可用的
     *      生产者的数据不丢失  ACK机制  producer是不断地往Kafka中写入数据，写入数据会有一个返回结果，表示是否写入成功。这里对应有一个ACKs的配置。
     *                      acks = 0：生产者只管写入，不管是否写入成功，可能会数据丢失。性能是最好的
     *                      acks = 1：生产者会等到leader分区写入成功后，返回成功，接着发送下一条
     *                      acks = -1/all：确保消息写入到leader分区、还确保消息写入到对应副本都成功后，接着发送下一条，性能是最差的
     *                      根据业务情况来选择ack机制，是要求性能最高，一部分数据丢失影响不大，可以选择0/1。如果要求数据一定不能丢失，就得配置为-1/all。
     *      消费者的数据不丢失
     *                      在消费者消费数据的时候，只要每个消费者记录好oﬀset值即可，就能保证数据不丢失。
     */

    /**
      联动中心:
        联动列表:新增
            选择设备类型:
                 根据tag标签 来区分每种设备 这里我是给前端返回了一个树形的结构图 比如:环境下面有温湿度设备、co2设备,根据温湿度设备又区分温度还是湿度
            设置联动条件
                到温度这里我们根据单位来设置条件,比如 温度 可能是 20°、30°,但是门可能是 开,关 这里做了个判断展示不同的效果
            设置执行动作和报警级别,和报警内容
                分为4大类:远程开启设备、调用附近摄像头、推送短信、播放报警音频
                        反控设备我是保存了 反控设备的设备id和命令
                        摄像头是保存了流地址
            触发联动条件
                触发联动也是监控kafka的topic
                先查询redis中缓存的告警条件,这里用了一个Spring的定时器 每隔5分钟会CachePut下缓存
                这里就一个简单的判断方法 通过比较 缓存 和kafka的数据来保存告警信息
                保存告警信息后 会去调用job-server中的反控协议 进行反控 这里我调用了同事写的feign接口 并且保存 反控的结果
                之后给上游推送告警信息,视频流地址,和音频秒数
                短信没用,反正给我的设计图是有短信,但是讨论的时候阉割掉了
            保存联动信息
        联动列表:查询
            还有个联动条件的查询、 有批量开启,关闭,删除 和根据类型、报警内容、级别、触发动作 、开启关闭的一个分页查询
        联动日志:查询
            根据类型,年月日进行区间查询
     */

    /**
     *  报警中心:
     *      报警日志:
     *         根据类型、内容、级别、处理状态、时间区间的一个查询
     *         主要展示 设备、告警类型、级别、开始时间 和结束时间
     *         结束时间是通过消费kafka 进行判断的时候如果不满足会形成一个闭环
     *         处理告警、处理人、时间、内容的一个详情查询
     *         还有定位、通过告警信息定位到3维平台的设备上
     *       右上角报警信息数量
     *          通过右上角可以直接跳转到告警中心
     *          离开和进入这个页面 都会清空缓存里面的告警信息数量
     */

    /**
     * 中钢项目
     *  能源管理:
     *      能耗展示:
     *          分项能耗:
     *              能耗主要分水和电
     *               通过时间统计这个时间段内 所有的水/电,按类型分类 返回一个列表 前端用 ECharts 展示的圆饼图
     *               这里我们是有统计平台,通过XXLjob 定时 每小时,每天,每月,每年统计
     *               我这统计平台的基础上在整合在一起把数据进行处理
     *          分时能耗
     *              分时是把具体的空调用电、电梯用电之类的 按时间段来统计 返回一个 ECharts 条形图来查看 每日的用量 并且返回10天用量最多的数据
     *          用能统计
     *              根据空间、水电类型、具体到某个具体的设备 按日期查询这个设备的使用量
     *              这里没什么技术点就多个平台通过feign接口 来回调用 处理数据
     *      统计分析:
     *          空间对比:
     *              根据空间查询多个楼层的水电 返回一个 ECharts 条形图 和一个列表
     *          分类对吧:
     *              根据类型对比损耗的详细信息
     *
     *
     *
     */


    /**
     * 安全运营时间:  通过回去taos库第一条数据 来判断
     *
     * 停车列表、车辆数量
     *  停车数据属于时间数据 存储在Mongodb中 ,这里有个麻烦的点就是 存储的数据不是很全 比如具体的车辆位置,这个在mysql库里 ,所以这里处理数据有点麻烦,并且连接mongodb的库是基础服务
     *  不能设计到业务,这里是写了有个通用的mongodb接口 然后在上游平台 3D里面再通过调用设备平台来把两种数据进行处理,然后通过3D的websocket来通知前台刷新接口实现数据更新
     * 门禁出入记录
     *  这个展示的数据挺全 基本上 直接调用通用的MG接口就行
     *  电梯推送上下行,推送电梯楼层
     *  这里是通过前台订阅传感器 我在3D平台进行推送最新的信息 ,前台进行替换
     *
     */


    @KafkaListener(groupId = KafkaConstants.TOPIC_WYZ, topics = KafkaConstants.WYZ,id = "wyz-2")
    public void process2(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {

        try {
            System.out.println(KafkaConstants.TOPIC_WYZ+">>>>>>>>>>>>topic = {"+ record.topic()+"}, offset = {"+ record.offset()+"}, value = {"+record.value()+"} ");
            acknowledgment.acknowledge();
        }catch (Exception e){
            e.printStackTrace();
        }

    }
}
